package com.atguigu.day03;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

public class Flink01_Source_Custom {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //TODO 2.从自定义的数据源获取数据（自定义Source）
        DataStreamSource<WaterSensor> streamSource = env.addSource(new MySource()).setParallelism(2);

        streamSource.print();

        env.execute();
    }

//    public static class MySource implements SourceFunction<WaterSensor>{

    //如果要设置多并行度，则需要去实现多并行度接口
    public static class MySource implements ParallelSourceFunction<WaterSensor>{


//        private Random random = new Random();
        private Boolean isRunning = true;

        /**
         * 生成数据
         * @param ctx
         * @throws Exception
         */
        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            while (isRunning){
                ctx.collect(new WaterSensor("sensor"+new Random().nextInt(100), System.currentTimeMillis(), new Random().nextInt(1000)));
                Thread.sleep(1000);
            }
        }

        /**
         * 取消生成数据
         * 一般在run方法中会有个while循环，通过此方法终止while循环
         */
        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
